package com.ds.infrastructure.hyperspace.container.flow.handler.zone;

import com.ds.infrastructure.hyperspace.common.Container;
import com.ds.infrastructure.hyperspace.common.flow.FlowHandlerPipeline;
import com.ds.infrastructure.hyperspace.container.flow.FlowHandlerInitException;
import com.ds.infrastructure.hyperspace.container.flow.context.HyperspaceDuplexFlowContext;
import com.ds.infrastructure.hyperspace.container.flow.handler.AbstractPreFlowHandler;
import com.ds.infrastructure.hyperspace.container.internal.operator.OperatorDefinition;
import com.ds.infrastructure.hyperspace.extension.HttpDataExtractor;
import com.ds.infrastructure.hyperspace.extension.HttpInputData;
import com.ds.infrastructure.hyperspace.extension.ShardingInfoVO;
import com.ds.infrastructure.zonesdk.Zone;
import io.netty.buffer.ByteBufUtil;
import io.netty.handler.codec.http.FullHttpRequest;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * usage eg:
 *  -   handler: customHttpExtensionZonePreHandler
 *      handlerParams: [bj, extensionId, uri|headerName, (operatorName,operatorArgs)...]
 * @author  Weishaoying
 */
@Slf4j
public class CustomHttpExtensionZonePreFlowHandler extends AbstractPreFlowHandler {

    private static final int INPUT_NAME_INDEX = 2;

    private volatile String defaultZone;

    private volatile String inputName;

    private volatile HttpDataExtractor httpDataExtractor;

    private volatile List<OperatorDefinition> operatorDefinitionList;

    @Override
    public void handleInput(HyperspaceDuplexFlowContext duplexFlowContext, Object data) {
        if (data instanceof FullHttpRequest) {
            if (inputName != null && extractAndComputeInput(duplexFlowContext,
                    inputName, operatorDefinitionList) == null) {
                assertZoneIsActive(defaultZone, null, duplexFlowContext, null, null);
                return;
            }
            if (httpDataExtractor == null) {
                return;
            }
            ShardingInfoVO shardingInfoVO = extractShardingInfo((FullHttpRequest)data);
            String shardingType = null;
            String shardingKey = null;
            Zone zone = null;
            if (shardingInfoVO != null && !StringUtils.isEmpty(shardingInfoVO.getShardingKey())) {
                shardingKey = shardingInfoVO.getShardingKey();
                shardingType = shardingInfoVO.getShardingKeyType();
                zone = findZone(duplexFlowContext, shardingType, shardingKey);
            }
            assertZoneIsActive(defaultZone, zone, duplexFlowContext, shardingType, shardingKey);
        }
    }

    @Override
    public String handlerName() {
        return "customHttpExtensionZonePreHandler";
    }

    @Override
    public void initHandlerParameters(Container container, FlowHandlerPipeline pipeline, String[] handlerParams) {
        if (!StringUtils.isEmpty(handlerParams[0])) {
            defaultZone = handlerParams[0];
        }
        String extensionId = handlerParams[1];
        if (StringUtils.isEmpty(extensionId)) {
            log.warn("Custom extension id is null!");
            throw new FlowHandlerInitException("Custom extension id not found!");
        }
        httpDataExtractor = (HttpDataExtractor)container.getDataExtractor(extensionId);
        if (httpDataExtractor == null) {
            log.warn("Custom extension corresponding HttpDataExtractor not found, id {}.", extensionId);
            throw new FlowHandlerInitException("Custom extension id invalid!");
        }
        if (handlerParams.length > INPUT_NAME_INDEX) {
            inputName = handlerParams[INPUT_NAME_INDEX];
            operatorDefinitionList = parseOperatorDefinition(container, handlerParams, INPUT_NAME_INDEX + 1);
        }
    }

    private ShardingInfoVO extractShardingInfo(FullHttpRequest fullHttpRequest) {
        Map<String, String> headerMap = new HashMap<>(16);
        Iterator<Map.Entry<String, String>> httpHeaders = fullHttpRequest.headers().iteratorAsString();
        while (httpHeaders.hasNext()) {
            Map.Entry<String, String> headerEntry = httpHeaders.next();
            headerMap.put(headerEntry.getKey(), headerEntry.getValue());
        }
        return httpDataExtractor.extract(
                new HttpInputData(fullHttpRequest.uri(), headerMap,
                        ByteBufUtil.getBytes(fullHttpRequest.content())));
    }
}
